package com.gukun.springboot.tdengine.service.highvolume;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.RandomUtil;
import com.gukun.springboot.tdengine.pojo.TagInfo;
import com.gukun.springboot.tdengine.domain.enums.TagType;
import com.gukun.springboot.tdengine.service.highvolume.pojo.DemoTagData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;

class ReadTask implements Runnable {
    private final static Logger logger = LoggerFactory.getLogger(ReadTask.class);
    private final int taskId;
    private final List<BlockingQueue<String>> taskQueues;
    private final int queueCount;
    private final int tableCount;
    private boolean active = true;

    public ReadTask(int readTaskId, List<BlockingQueue<String>> queues, int tableCount) {
        this.taskId = readTaskId;
        this.taskQueues = queues;
        this.queueCount = queues.size();
        this.tableCount = tableCount;
    }

    /**
     * Assign data received to different queues.
     * Here we use the suffix number in table name.
     * You are expected to define your own rule in practice.
     *
     * @param index record received
     * @return which queue to use
     */
    public int getQueueId(int  index) {
        return index % this.queueCount;
    }

    @Override
    public void run() {
        logger.info("started");
        List<TagInfo> tagInfos = DataUtil.getTagInfos(100000);
        List<DemoTagData> tagDataList = new ArrayList<>();
        for (TagInfo tagInfo :
                tagInfos) {
            DemoTagData tagData = new DemoTagData(tagInfo.getTagName(),System.currentTimeMillis(), "0",(byte) 0);
            switch (tagInfo.getTagType()) {
                case TagType.DOUBLE:
                    tagData.setV(RandomUtil.randomDouble() + "");
                    break;
                case TagType.FLOAT:
                    tagData.setV(RandomUtil.getRandom().nextFloat() + "");
                    break;
                case TagType.INT:
                    tagData.setV(RandomUtil.getRandom().nextInt() + "");
                    break;
                case TagType.BOOL:
                    tagData.setV(RandomUtil.getRandom().nextBoolean() + "");
                    break;
            }

            tagDataList.add(tagData);
        }


        MockDataSource it = new MockDataSource(tagDataList);
        try {
            int index = 0;
            DateTime lastDt = DateTime.now();

            while (it.hasNext() && active) {
                DateTime now = DateTime.now();
                if(DateUtil.between(lastDt,now, DateUnit.MS) > 1000){
                    //大于1秒，就等吧
                    lastDt =  DateTime.now();
                }else{
                    index ++;
                    String line = it.nextDate(lastDt);
                    if(line == null) {
                        Thread.sleep(100);
                        continue;
                    }
                    int queueId = getQueueId(index);
                    taskQueues.get(queueId).put(line);
                }
            }
        } catch (Exception e) {
            logger.error("Read Task Error", e);
        }
    }

    public void stop() {
        logger.info("stop");
        this.active = false;
    }
}